package org.study.async;

import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 作用描述
 *
 * @author doveylovey
 * @version v1.0.0
 * @email 1135782208@qq.com
 * @date 2024年08月08日
 */
@Slf4j
public class CompletableFutureTest05 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        testSync();
        testAsync();
        testCompletableFuture();
        queryAsync();
    }

    /**
     * 同步任务
     */
    public static void testSync() {
        long start = System.currentTimeMillis();
        int A = taskA();
        int B = taskB(2);
        int C = taskC();
        int D = taskD();
        int E = taskE();
        log.info("同步任务总用时：{}毫秒，执行结果为：{}", System.currentTimeMillis() - start, A + B + C + D + E);
    }

    /**
     * 异步任务
     */
    public static void testAsync() throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        // 创建个核心线程7，最大线程7，等待队列为5，解决策略为抛出异常的固定线程池。
        ExecutorService executorService = new ThreadPoolExecutor(7, 7, 0L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
        // 执行异步任务
        Future<Integer> tackA = executorService.submit(() -> taskA());
        Future<Integer> tackB = executorService.submit(() -> taskB(2));
        Future<Integer> tackC = executorService.submit(() -> taskC());
        Future<Integer> tackD = executorService.submit(() -> taskD());
        Future<Integer> tackE = executorService.submit(() -> taskE());

        // 上面的任务都是异步化执行的，最后对执行结果进行汇总合并
        Integer A = tackA.get();
        Integer B = tackB.get();
        Integer C = tackC.get();
        Integer D = tackD.get();
        Integer E = tackE.get();
        log.info("异步任务总用时：{}毫秒，执行结果为：{}", System.currentTimeMillis() - start, A + B + C + D + E);
    }

    /**
     * 有五个任务：
     * 任务A：有返回值
     * 任务B：要依赖任务A返回的结果，并且有返回值
     * 任务C：独立，有返回值
     * 任务D：和E的执行完成后进行操作，并有返回值
     * 最后三个任务结果相加得到最后结果返回
     */
    public static void testCompletableFuture() throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        // 创建一个线程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(20, 20, 0L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
        // 异步任务AB
        CompletableFuture<Integer> futureAB = CompletableFuture.supplyAsync(() -> {
            return taskA();
        }, threadPoolExecutor).thenApplyAsync((res) -> {
            // 传入的res是taskA任务执行的返回结果
            return taskB(res);
        });
        // 异步任务C
        CompletableFuture<Integer> futureC = CompletableFuture.supplyAsync(() -> {
            return taskC();
        }, threadPoolExecutor);
        // 异步任务DE
        CompletableFuture<Integer> futureDE = CompletableFuture.supplyAsync(() -> {
            return taskD();
        }, threadPoolExecutor).thenCombineAsync(CompletableFuture.supplyAsync(() -> {
            return taskE();
        }, threadPoolExecutor), (resD, ResE) -> {
            // 任务DE的结果相加
            return resD + ResE;
        });
        Integer AB = futureAB.get();
        Integer C = futureC.get();
        Integer DE = futureDE.get();
        log.info("异步任务总用时：{}毫秒，执行结果为：{}", System.currentTimeMillis() - start, AB + C + DE);
    }

    /**
     * 异步返回结果
     */
    public static void queryAsync() {
        log.info("查询数据 queryAsync -- 开始");
        Map<String, Object> map = new HashMap<>();
        // 1、查询数据一
        CompletableFuture<String> twentyFourQuery = CompletableFuture.supplyAsync(() -> {
            log.info("查询数据1 -- 开始");
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info("查询数据1 -- 结束");
            return "数据一";
        });
        twentyFourQuery.thenAccept((result) -> {
            map.put("data1", result);
            if (RandomUtil.randomBoolean()) {
                // 随机模拟异常
                throw new RuntimeException("模拟异常");
            }
        }).exceptionally((e) -> {
            log.error("查询数据一异常：{}", e.getMessage(), e);
            map.put("data1", "异常数据一");
            return null;
        });
        // 2、查询数据二
        CompletableFuture<String> constellationQuery = CompletableFuture.supplyAsync(() -> {
            log.info("查询数据2 -- 开始");
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info("查询数据2 -- 结束");
            return "数据二";
        });
        constellationQuery.thenAccept((result) -> {
            map.put("data2", result);
        }).exceptionally((e) -> {
            log.error("查询数据二异常：{}", e.getMessage(), e);
            map.put("data2", "异常数据二");
            return null;
        });
        // 3、allOf-两个查询必须都完成
        CompletableFuture<Void> allQuery = CompletableFuture.allOf(twentyFourQuery, constellationQuery);
        CompletableFuture<Map<String, Object>> future = allQuery.thenApply((result) -> {
            log.info("------------------ 全部查询都完成 ------------------ ");
            return map;
        }).exceptionally((e) -> {
            log.error(e.getMessage(), e);
            return null;
        });
        // 获取异步方法返回值：get()方法内部抛出了异常需手动处理；join()方法内部处理了异常无需手动处理，点进去一看便知
        future.join();
        log.info("查询数据 queryAsync -- 结束");
        log.info("结果：{}", JSON.toJSONString(map));
    }

    public static int taskA() {
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 2;
    }

    public static int taskB(int param) {
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return param;
    }

    public static int taskC() {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 1;
    }

    public static int taskD() {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 1;
    }

    public static int taskE() {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 1;
    }
}
